/*
 *  Copyright 2020-2025 the original author or authors.
 *  You cannot use this file unless authorized by the author.
 */

package org.ipig.computing.spark.executer;

import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.ipig.commons.helper.AssertHelper;
import org.ipig.commons.service.cmd.CommandService;
import org.ipig.computing.constant.context.SparkContext;
import org.ipig.computing.spark.conf.ApplicationConf;
import org.ipig.constants.JavaCnst;
import org.ipig.constants.ResultStatus;
import org.ipig.constants.SymbolCnst;

import java.io.BufferedInputStream;
import java.io.BufferedReader;
import java.io.File;
import java.io.InputStreamReader;
import java.util.concurrent.TimeUnit;

/**
 * SparkLinuxCommandExecuter
 *
 * @author <a href="mailto:comchnts@163.com">chinats</a>
 * @since 1.0
 */
@Slf4j
@Data
public class SparkLinuxCommandExecuter implements CommandService<ApplicationConf, ResultStatus> {
    private long timeout=30000;
    @Override
    public ResultStatus execute(ApplicationConf conf) {
        ResultStatus resultStatus = ResultStatus.SUCCESS;
        String command = toCommand(conf);
        try {
            //ProcessBuilder.start()创建一个本机进程，并返回一个Process子类的一个实例，该实例可以获取进程的相关信息，也可以控制进程。
            Process process = Runtime.getRuntime().exec(command);
            //##读取命令的输出信息,对于io流如果有缓冲大小限制，则可能出现阻塞，导致死锁情况。
            BufferedInputStream bis = new BufferedInputStream(process.getInputStream());
            BufferedReader reader = new BufferedReader(new InputStreamReader(bis));
            boolean flag = process.waitFor(timeout, TimeUnit.MILLISECONDS);
            if (process.exitValue() != 0) {
                log.warn("执行失败,Command={}", command);
            }
            String s = null;
            while ((s = reader.readLine()) != null) {
                log.info(s);
            }
            bis.close();
            reader.close();
        } catch (Exception e) {
            log.error(ExceptionUtils.getStackTrace(e));
            e.printStackTrace();
            resultStatus = ResultStatus.FAILURE;
        }
        return resultStatus;
    }


    @Override
    public String toCommand(ApplicationConf conf) {
        String javaHome = System.getenv(JavaCnst.ENV_JAVA_HOME);
        AssertHelper.hasText(javaHome, "Set the ENV_JAVA_HOME environment variable!");
        String javaCmd = javaHome + File.separator + "bin" + File.separator + "java";
        String sparkHome = System.getenv("SPARK_HOME");
        AssertHelper.hasText(sparkHome, "Set the SPARK_HOME environment variable!");
        String sparkLibDir = sparkHome + File.separator + "jars" + File.separator + SymbolCnst.ASTERISK;
        String sparkConfDir = sparkHome + File.separator + "conf" + File.separator;
        String hadoopHome = System.getenv("HADOOP_HOME");
        AssertHelper.hasText(hadoopHome, "Set the HADOOP_HOME environment variable!");
        String hadoopConfDir = hadoopHome + File.separator + "etc" + File.separator + "hadoop" + File.separator;
        //以上参数动态获取
        String appLibDir = conf.getAppLibDir();
        String mainClass = conf.getMainClass();
        String mainArgs = conf.getMainArg();
        String appName = conf.getAppName();
        String appResource = conf.getAppResource();
        String deployMode = SparkContext.DeployMode.parse(conf.getAppDeployMode()).code;
        String workMode = SparkContext.WorkingMode.YARN.defaultValue;
        AssertHelper.hasText(appLibDir, "please enter the local path of the application dependency package【appLibDir】");
        int index = StringUtils.lastIndexOf(appLibDir, File.separator);
        if (index > 0) {
            String endStr = StringUtils.substring(appLibDir, index + 1);
            if (StringUtils.isBlank(endStr)) {
                appLibDir = appLibDir + SymbolCnst.ASTERISK;
            } else {
                int locIndex = StringUtils.indexOfAny(endStr, SymbolCnst.ASTERISK);
                if (locIndex < 0) {
                    appLibDir = appLibDir + File.separator + SymbolCnst.ASTERISK;
                }
            }
        }
        AssertHelper.hasText(mainClass, "enter the fully qualified path of the main function class【mainClass】");
        AssertHelper.hasText(appResource, "enter the fully qualified path of the package where the main function class is located【appResource】");
        StringBuilder command = new StringBuilder();
        command.append(javaCmd).append(JavaCnst.Cmd.CP).append(sparkConfDir)
                .append(SymbolCnst.COLON).append(hadoopConfDir);
        if (StringUtils.isNotBlank(conf.getAppConfDir())) {
            command.append(SymbolCnst.COLON).append(conf.getAppConfDir());
        }
        command.append(SymbolCnst.COLON).append(sparkLibDir)
                .append(SymbolCnst.COLON).append(appLibDir)
                .append(SymbolCnst.SPACE).append(SparkContext.SPARK_SUBMIT_CLASS).append(SymbolCnst.SPACE)
                .append(SparkContext.APP_CLASS).append(mainClass).append(SymbolCnst.SPACE);
        if (StringUtils.isNotBlank(appName)) {
            command.append(SparkContext.APP_NAME).append(appName);
        }
        command.append(SparkContext.APP_MASTER).append(workMode).append(SparkContext.APP_DEPLOY_MODE).append(deployMode).append(SymbolCnst.SPACE).append(appResource);
        if (StringUtils.isNotBlank(mainArgs)) {
            command.append(SymbolCnst.SPACE).append(mainArgs);
        }
        return command.toString();
    }
}
